package com.cnebula.dataprocess.task.center.scheduler;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import javax.annotation.Resource;

import org.apache.rocketmq.spring.starter.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import com.cnebula.dataprocess.common.JobPayload;
import com.cnebula.dataprocess.common.JobResult;
import com.cnebula.dataprocess.common.JobStatus;
import com.cnebula.dataprocess.common.hive.HiveDataPass;
import com.cnebula.dataprocess.common.hive.HiveLoadData;
import com.cnebula.dataprocess.common.hive.HiveLoadDataJobResult;
import com.cnebula.dataprocess.common.hive.HivePurge;
import com.cnebula.dataprocess.common.hive.HivePurgeJobResult;
import com.cnebula.dataprocess.common.sqoop.DataSourceNode;
import com.cnebula.dataprocess.common.sqoop.SqoopJobResult;
import com.cnebula.dataprocess.task.center.MainTask;
import com.cnebula.dataprocess.task.center.MainTaskResult;
import com.cnebula.dataprocess.task.center.Task;
import com.cnebula.dataprocess.task.center.TaskResult;
import com.cnebula.dataprocess.task.center.TaskStatus;
import com.google.gson.Gson;

@Component
public class TaskScheduler {

	private static final Logger log = LoggerFactory.getLogger(TaskScheduler.class);
	private Gson gson = new Gson();

	private final ConcurrentHashMap<String, Task> taskMap = new ConcurrentHashMap<String, Task>();

	private final ConcurrentHashMap<String, MainTask> mainTaskMap = new ConcurrentHashMap<String, MainTask>();

	@Resource
	private RocketMQTemplate rocketMQTemplate;

	public void startTask(Task task) {

		LinkedList<JobPayload> jobs = task.getJobs();
		JobPayload payload = jobs.poll();
		// TODO 目前 启动入口的作业都是sqoop job
		DataSourceNode node = (DataSourceNode) payload;

		rocketMQTemplate.convertAndSend("datanode-topic", node);
	}

	public void completedTask(Task task) {
		String mainTaskId = task.getMainTaskId();
		String taskId = task.getTaskId();

		taskMap.remove(taskId);

		MainTask mainTask = mainTaskMap.get(mainTaskId);
		AtomicInteger counter = mainTask.getCounter();
		List<Task> tasks = mainTask.getTasks();
		int n = counter.incrementAndGet();

		if (n == tasks.size()) {

			MainTaskResult mainTaskRst = getMainTaskResult(tasks);
			mainTaskRst.setId(mainTaskId);
			mainTaskRst.setMessage("");

			// main task 完成
			log.info("=============send main task {} complete message=================================", mainTaskRst.getId());
			log.info(gson.toJson(mainTaskRst));
			rocketMQTemplate.convertAndSend("hive_data_process_new", mainTaskRst);
			log.info("================================================================================");

			mainTaskMap.remove(mainTaskId);
		}
	}

	private MainTaskResult getMainTaskResult(List<Task> tasks) {
		MainTaskResult rst = new MainTaskResult();
		List<TaskResult> taskResults = new ArrayList<TaskResult>();
		for (Task t : tasks) {
			TaskResult taskRst = new TaskResult();
			taskRst.setId(t.getTaskId());
			taskRst.setStatus(t.getStatus());
			taskRst.setMessage(t.getMessage());

			taskResults.add(taskRst);
		}
		rst.setTaskResults(taskResults);
		rst.setStatus(getMainTaskStatus(tasks));

		return rst;
	}

	private int getMainTaskStatus(List<Task> tasks) {

		int status = 400;

		// check all ok
		boolean isOK = true;
		for (Task t : tasks) {
			if (t.getStatus() != TaskStatus.OK) {
				isOK = false;
				break;
			}
		}

		if (isOK) {
			return 200;
		}

		// check all fail
		boolean isFail = true;
		for (Task t : tasks) {
			if (t.getStatus() != TaskStatus.FAIL) {
				isFail = false;
				break;
			}
		}

		if (isFail) {
			return 500;
		}

		return status;

	}

	public void submitMainTask(MainTask mainTask) {
		mainTaskMap.put(mainTask.getId(), mainTask);
		List<Task> tasks = mainTask.getTasks();
		submitTasks(tasks);
	}

	private void submitTasks(List<Task> tasks) {
		for (Task task : tasks) {
			task.setStatus(TaskStatus.RUNNING);
			taskMap.put(task.getTaskId(), task);
			startTask(task);
		}
	}

	// public void submitTask() {
	//
	// Task task = new Task();
	// task.setStatus(TaskStatus.RUNNING);
	// task.setTaskId(UUID.randomUUID().toString());
	//
	// DataSourceNode node = new DataSourceNode();
	// node.setHost("222.29.81.136");
	// node.setPort(3306);
	// node.setCharset("utf-8");
	// node.setDbType("mysql");
	// node.setDatabase("_510500");
	// node.setUsername("root");
	// node.setPassword("calis123");
	// node.setSql("select * from test_goods");
	// node.setPartition("id");
	// node.setJobId(UUID.randomUUID().toString());
	// node.setTaskId(task.getTaskId());
	//
	// HiveLoadData hiveLoadData = new HiveLoadData();
	// hiveLoadData.setJobId(UUID.randomUUID().toString());
	// hiveLoadData.setTaskId(task.getTaskId());
	// hiveLoadData.setHiveTable("test_goods");
	//
	// HivePurge hivePurge = new HivePurge();
	// hivePurge.setJobId(UUID.randomUUID().toString());
	// hivePurge.setTaskId(task.getTaskId());
	// hivePurge.setSql("insert into goods select id, name, price, price_str,
	// iscollect from test_goods");
	//
	// HiveDataPass hiveDataPass = new HiveDataPass();
	// hiveDataPass.setJobId(UUID.randomUUID().toString());
	// hiveDataPass.setTaskId(task.getTaskId());
	// hiveDataPass.setPassColumn("id");
	// hiveDataPass.setTable("goods");
	//
	//
	// LinkedList<JobPayload> jobs = new LinkedList<JobPayload>();
	// jobs.add(node);
	// jobs.add(hiveLoadData);
	// jobs.add(hivePurge);
	// jobs.add(hiveDataPass);
	//
	// task.setJobs(jobs);
	//
	// taskMap.put(task.getTaskId(), task);
	// startTask(task);
	// }

	public void next(JobResult result) {
		if(result == null){
			log.warn("job result is null");
			return;
		}
		Task task = taskMap.get(result.getTaskId());
		if(task == null){
			log.warn("the task {} has been discarded", result.getTaskId());
			return;
		}
		LinkedList<JobPayload> jobs = task.getJobs();
		JobPayload payload = jobs.poll();
		try {

			if (result.getStatus() != JobStatus.OK) {
				task.setStatus(TaskStatus.FAIL);
				task.setMessage(result.getMessage());
				this.completedTask(task);
				return;
			}

			if (payload == null) {
				task.setStatus(TaskStatus.OK);
				this.completedTask(task);
				return;
			}

			if (result.getClass().getName().equals(SqoopJobResult.class.getName())) {
				SqoopJobResult rst = (SqoopJobResult) result;
				String datafile = rst.getDatafile();
				HiveLoadData hiveLoadData = (HiveLoadData) payload;
				hiveLoadData.setDatafile(datafile);
				rocketMQTemplate.convertAndSend("hive-loaddata-topic", hiveLoadData);
			}

			if (result.getClass().getName().equals(HiveLoadDataJobResult.class.getName())) {
				if (payload.getClass().getName().equals(HivePurge.class.getName())) {
					HivePurge hivePurge = (HivePurge) payload;
					rocketMQTemplate.convertAndSend("hive-purge-topic", hivePurge);
				} else if (payload.getClass().getName().equals(HiveDataPass.class.getName())) {
					HiveDataPass hiveDataPass = (HiveDataPass) payload;
					rocketMQTemplate.convertAndSend("hive-datapass-topic", hiveDataPass);
				}
			}

			if (result.getClass().getName().equals(HivePurgeJobResult.class.getName())) {
				if (payload.getClass().getName().equals(HiveDataPass.class.getName())) {
					HiveDataPass hiveDataPass = (HiveDataPass) payload;
					rocketMQTemplate.convertAndSend("hive-datapass-topic", hiveDataPass);
				}
			}
		} catch (Throwable e) {
			task.setStatus(TaskStatus.FAIL);
			task.setMessage(e.getMessage());
			this.completedTask(task);
		}
	}
}
